package service

import (
	"context"
	"errors"
	"sync"
	"time"

	"hundsun.com/hsl/hschain/common/util"

	lru "github.com/hashicorp/golang-lru"
	"hundsun.com/hsl/hschain/common/log/logging"
	pbcom "hundsun.com/hsl/hschain/protos/common"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
	protosrpc "hundsun.com/hsl/hschain/protos/rpc"
	"hundsun.com/hsl/hschain/rpc/common/constant"
	"hundsun.com/hsl/hschain/rpc/common/utils"
	dbm "hundsun.com/hsl/hschain/store/db"
	"hundsun.com/hsl/hschain/store/mq"
)

//StreamEnhance StreamEnhance
type StreamEnhance struct {
	protosrpc.Rpc_ProcessSubRequestServer
	Disable   int32
	SleepTime int32
}

//EndPointMonitor EndPointMonitor
type EndPointMonitor struct {
	lock         sync.Mutex
	endPointMap  map[string]*RequestMonitor
	ctx          context.Context
	consumeBlk   chan uint64
	workerClient mq.Client
	nextBlockNum uint64
	Log          logging.Log
	db           dbm.Database
	limit        int
}

//NewEndPointMonitor 新建
func NewEndPointMonitor(ctx context.Context, log logging.Log, workerClient mq.Client, db dbm.Database) *EndPointMonitor {
	monitor := &EndPointMonitor{
		ctx:          ctx,
		Log:          log,
		consumeBlk:   make(chan uint64, 5),
		workerClient: workerClient,
		db:           db,
		limit:        constant.EndPointNumberLimit,
		endPointMap:  make(map[string]*RequestMonitor),
	}
	monitor.InitNextBlockNum()
	go monitor.refreshMonitor()
	return monitor
}

//SetWorkerClient 设置workerClient
func (e *EndPointMonitor) SetWorkerClient(wClient mq.Client) {
	e.Log.Infof("update worker client")
	e.workerClient = wClient
}

//GetWorkerClient 获取workerClient
func (e *EndPointMonitor) GetWorkerClient() mq.Client {
	return e.workerClient
}

//InitNextBlockNum nextBlockNum初始化
func (e *EndPointMonitor) InitNextBlockNum() {

	buf, _ := e.db.Get([]byte(dbm.NamespaceRPCBlkNum + "blockNumber"))
	blkNum, _ := util.Bytes2Struct(uint64(0), buf)
	if blkNum == nil {
		e.nextBlockNum = 1
	} else {
		e.nextBlockNum = blkNum.(uint64)
	}
	e.Log.Infof("nextBlockNumber initialized, blockNumber:%d", e.nextBlockNum)
}

//Consume 最新区块处理协程
func (e *EndPointMonitor) Consume(msg *mq.Message) {

	subRes := msg.Data.(*pbcom.RspSubscriber)
	headerI, err := util.Bytes2Struct(&pbledger.BlockHeader{}, subRes.GetNormal().Data)
	if err != nil {
		return
	}
	header := headerI.(*pbledger.BlockHeader)
	e.Log.Infof("receive block executed message, blockNumber:%d", header.Number)
	e.consumeBlk <- header.Number
}

func (e *EndPointMonitor) refreshMonitor() {
	for {
		select {
		case <-e.ctx.Done():
			e.Log.Infof("refresh monitor stopped")
			return
		case num := <-e.consumeBlk:
			e.refreshNextBlock(num)
		}
	}
}

func (e *EndPointMonitor) refreshNextBlock(blkNum uint64) {
	//mq中的区块小于等于当前区块
	if blkNum < e.nextBlockNum {
		e.Log.Warnf("block number in mq is not latest, targetBlockNum:%d, latestBlockNum:%d", blkNum, e.nextBlockNum)
		return
	}
	for {
		if e.nextBlockNum == blkNum+1 {
			e.Log.Infof("rpc nextBlockNum updated to %d", e.nextBlockNum)
			buf, _ := util.Struct2Bytes(e.nextBlockNum)
			err := e.db.Set([]byte(dbm.NamespaceRPCBlkNum+"blockNumber"), buf)
			if err != nil {
				e.Log.Errorf("rpc store blockNumber failed, reason:%v", err)
			}
			return
		}
		block, err := GetExecutedBlock(e.workerClient, e.nextBlockNum)
		if err != nil {
			e.Log.Error(err)
		}
		if err != nil || block == nil {
			time.Sleep(time.Millisecond * 200)
			e.Log.Errorf("executed block is nil, blockNumber:%d", e.nextBlockNum)
			continue
		}
		//解析区块，向所有subscribeMonitor发送订阅结果
		e.Log.Infof("rpc endpoint num:%d", len(e.endPointMap))
		for _, requestMonitor := range e.endPointMap {
			go requestMonitor.ResolveBlock(block)
		}
		e.nextBlockNum++
	}
}

//NewEndpoint 新增endpoint或更新原endpoint stream
func (e *EndPointMonitor) NewEndpoint(stream *StreamEnhance) (*RequestMonitor, error) {
	ip, _ := utils.GetClientIP(stream.Context())
	if len(e.endPointMap) >= e.limit {
		return nil, errors.New("endpoint number exceed limit")
	}
	//判断endpoint是否已存在
	var monitor *RequestMonitor
	var ok bool
	if monitor, ok = e.Get(ip); !ok {
		monitor = NewMonitor(e, ip)
	}
	monitor.RefreshStream(stream)
	e.Set(ip, monitor)
	return monitor, nil
}

//Set 新增endpoint
func (e *EndPointMonitor) Set(endpoint string, request *RequestMonitor) {
	e.lock.Lock()
	defer e.lock.Unlock()
	e.endPointMap[endpoint] = request
}

//Get 获取endpoint
func (e *EndPointMonitor) Get(endpoint string) (*RequestMonitor, bool) {
	e.lock.Lock()
	defer e.lock.Unlock()
	v, ok := e.endPointMap[endpoint]
	return v, ok
}

//Delete 删除endpoint
func (e *EndPointMonitor) Delete(endpoint string) {
	e.lock.Lock()
	defer e.lock.Unlock()
	e.Log.Infof("endpoint deleted, ip:%s", endpoint)
	delete(e.endPointMap, endpoint)
}

//RequestMonitor RequestMonitor
type RequestMonitor struct {
	//requestMap仅内部使用
	EndPoint string
	//保存申请的map
	requestMap map[string]protosrpc.IBlockSubscribe
	//保存发送结果的map
	sendMap *lru.Cache
	//收到新块channel
	ResultsChan         chan *ResultsEnhanced
	CancelFun           context.CancelFunc
	monitorCtx          context.Context
	Stream              *StreamEnhance
	Parent              *EndPointMonitor
	SendNextBlockNumber uint64
	mpLock              sync.Mutex
	//timer      chan int32
}

//ResultsEnhanced ResultsEnhanced
type ResultsEnhanced struct {
	BlockNumber uint64
	SubResult   *protosrpc.SubscriptionResult
}

//NewMonitor 新建endpointMonitor
func NewMonitor(parent *EndPointMonitor, endpoint string) *RequestMonitor {
	//requestMonitor只能内部创建
	parent.Log.Infof("new endpoint created, ip:%s", endpoint)
	monitorCtx, cancelFun := context.WithCancel(parent.ctx)
	sendMap, _ := lru.New(constant.BlockCacheLimit)
	monitor := &RequestMonitor{
		requestMap:          make(map[string]protosrpc.IBlockSubscribe, constant.SubscribeInit),
		ResultsChan:         make(chan *ResultsEnhanced, 5),
		sendMap:             sendMap,
		CancelFun:           cancelFun,
		monitorCtx:          monitorCtx,
		EndPoint:            endpoint,
		Parent:              parent,
		SendNextBlockNumber: parent.nextBlockNum,
	}
	go monitor.SubscribeMonitor()
	go monitor.SendMonitor()
	return monitor
}

//RequestMapPut 增加客户端请求
func (r *RequestMonitor) RequestMapPut(key string, val protosrpc.IBlockSubscribe) {
	r.mpLock.Lock()
	defer r.mpLock.Unlock()
	r.requestMap[key] = val
}

//RequestMapDelete 删除客户端请求
func (r *RequestMonitor) RequestMapDelete(key string) {
	r.mpLock.Lock()
	defer r.mpLock.Unlock()
	delete(r.requestMap, key)
}

//RefreshStream 更新流
func (r *RequestMonitor) RefreshStream(newStream *StreamEnhance) {
	r.Stream = newStream
}

//ResolveBlock 解析区块
func (r *RequestMonitor) ResolveBlock(block *pbledger.ExecutedBlock) {
	var onceResult []*protosrpc.Result
	r.mpLock.Lock()
	for subID, subInfo := range r.requestMap {
		isIn, isEnd, result := subInfo.Filter(block, protosrpc.Result_REALTIME)
		if isIn {
			r.Parent.Log.Debugf("tx sub matched, subID:%s", subID)
			onceResult = append(onceResult, result...)
		}
		if isEnd {
			delete(r.requestMap, subID)
		}
	}
	r.mpLock.Unlock()
	//send channel 发送结果

	r.ResultsChan <- &ResultsEnhanced{
		BlockNumber: block.Block.Header.Number,
		SubResult:   &protosrpc.SubscriptionResult{Result: onceResult},
	}
}

//SendMonitor 客户端响应
func (r *RequestMonitor) SendMonitor() {
	r.Parent.Log.Infof("sender monitor started, endpoint:%s", r.EndPoint)
	for {
		select {
		case <-r.monitorCtx.Done():
			r.Parent.Log.Infof("sender monitor receive goroutine closed, endpoint:%s", r.EndPoint)
			return
		//收到块解析结果，保存在map中,如果结果为空也要插入空结果
		case result := <-r.ResultsChan:
			if result == nil {
				continue
			}
			sendLog := r.Parent.Log.WithField(logging.LogFieldBlockNumber, result.BlockNumber)
			sendLog.Infof("sender monitor cache block")
			r.sendMap.Add(result.BlockNumber, result)
			//如果缓存的区块过多，则重置SendNextBlockNum
			if r.sendMap.Len() > constant.BlockCacheLimit-2 {
				if key, _, ok := r.sendMap.GetOldest(); ok {
					r.SendNextBlockNumber = key.(uint64)
				}
			}
			//如果不存在下一块报异常
			if _, ok := r.sendMap.Get(r.SendNextBlockNumber); !ok {
				r.Parent.Log.WithField(logging.LogFieldBlockNumber, r.SendNextBlockNumber).WithField(logging.LogFieldRPCEndpoint, r.EndPoint).Warn("block num not sec")
			}
			//如果区块连续，将结果返回给sdk
			for {
				tempLog := r.Parent.Log.WithField(logging.LogFieldBlockNumber, r.SendNextBlockNumber).WithField(logging.LogFieldRPCEndpoint, r.EndPoint)
				if valI, ok := r.sendMap.Get(r.SendNextBlockNumber); ok {
					send := valI.(*ResultsEnhanced)
					var err error
					responseSize := len(send.SubResult.Result)
					if responseSize != 0 {
						err = (*r.Stream).Send(send.SubResult)
					}
					r.Stream.SleepTime = 0
					if err != nil {
						tempLog.Errorf("failed to send subscribe endpoint:%s, reason:%v", r.EndPoint, err)
						break
					} else {
						tempLog.Debugf("send subscribe result success, endpoint:%s, response num:%d", r.EndPoint, responseSize)
						r.sendMap.Remove(r.SendNextBlockNumber)
						r.SendNextBlockNumber++
					}
				} else {
					break
				}
			}
		}
	}
}

//SubscribeMonitor 订阅监控
func (r *RequestMonitor) SubscribeMonitor() {
	//return 时关闭流
	r.Parent.Log.Infof("subscribe monitor started, endpoint:%s", r.EndPoint)
	timer := time.NewTicker(time.Second * constant.SubscribeCleanTime)
	for {
		select {
		case <-r.monitorCtx.Done():
			r.Parent.Log.Infof("subscribe monitor receive goroutine closed, endpoint:%s", r.EndPoint)
			//删除endpoint
			r.Parent.Delete(r.EndPoint)
			timer.Stop()
			return
		//定时任务：判断过期的request，刷新requestMap
		case <-timer.C:
			//如果stream失效删除endpoint
			if r.Stream.Disable == constant.Abnormal {
				r.Parent.Log.Infof("endpoint:%s cannot be connected and removed", r.EndPoint)
				r.CancelFun()
			}
			//刷新request map
			r.mpLock.Lock()
			newMap := make(map[string]protosrpc.IBlockSubscribe, constant.SubscribeInit)
			for key, subInfo := range r.requestMap {
				if !subInfo.IsExpire() {
					newMap[key] = subInfo
				}
			}
			r.Parent.Log.Infof("present subNum:%d, endpoint:%s", len(newMap), r.EndPoint)
			r.requestMap = newMap
			r.mpLock.Unlock()
		}
	}
}
